package 生产者;

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;

import java.util.Properties;

/**
 *Kafka的幂等性操作旨在确保消息即使在网络异常或重试发送的情况下，也能保证只被处理一次，从而避免重复消息的问题。要启用Kafka生产者的幂等性功能，需要满足以下要求和配置：
 *
 * 1. 配置幂等性
 * 开启幂等性：首先，需要在生产者配置中设置enable.idempotence=true。这会自动启用一些内部机制来确保消息的唯一性和顺序性。
 * properties
 * 深色版本
 * props.put("enable.idempotence", "true");
 * 2. 设置合适的acks策略
 * 确认级别：为了保证幂等性，必须将acks配置设置为all（或者-1），这意味着生产者将等待所有同步副本(in-sync replicas)确认接收了消息后才认为发送成功。这是为了确保消息不会因为leader切换而丢失。
 * properties
 * 深色版本
 * props.put("acks", "all");
 * 3. 限制未完成请求数量
 * 单个连接上的最大未完成请求数：为了确保消息顺序，需要将max.in.flight.requests.per.connection设置为1或更小的值。默认情况下，当启用幂等性时，Kafka会自动将其设置为5以下，但为了绝对确保顺序性，通常建议设置为1。注意，这可能会影响吞吐量。
 * properties
 * 深色版本
 * props.put("max.in.flight.requests.per.connection", "1");
 * 注意事项
 * 事务ID：虽然不是直接与幂等性相关的配置，但如果你打算使用Kafka的事务特性（例如，实现“读已提交”的语义），你需要为生产者配置一个transactional.id。不过，对于仅启用幂等性的场景，这不是必需的。
 * 性能影响：由于幂等性和严格的顺序保障，可能会对消息发送的吞吐量和延迟产生一定影响。因此，在实际应用中需要根据具体的需求进行权衡。
 * 通过上述配置，可以有效利用Kafka的幂等性特性来避免因网络问题、重试机制等原因导致的消息重复问题。然而，需要注意的是，幂等性并不能解决所有类型的消息重复问题，例如消费者端的重复消费，这些问题可能需要其他策略来解决，比如基于业务逻辑的去重机制。
 */
public class KafkaRetry {

    public static void main(String[] args) {
        // 设置不同的acks值来演示其区别


        Properties props = new Properties();
        // 指定Kafka集群地址
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "114.251.235.19:9092");
        // 设置acks配置
        props.put(ProducerConfig.ACKS_CONFIG, "-1");
        // 消息序列化器
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
        //幂等性配置
        props.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, true);
        //幂等性配置
        props.put(ProducerConfig.RETRIES_CONFIG, 5);


        KafkaProducer<String, String> producer = new KafkaProducer<>(props);

        try {
            // 创建要发送的消息记录
            ProducerRecord<String, String> record = new ProducerRecord<>("topic_test", "key", "value");
            // 发送消息
            producer.send(record).get(); // 使用get()方法进行同步发送，便于示例展示
        } catch (Exception e) {

            e.printStackTrace();
        } finally {
            // 关闭生产者
            producer.close();
        }
    }

}